package com.yejiarong.influxdb.service;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.util.DateUtils;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.rabbitmq.client.Channel;
import com.yejiarong.common.constant.InfluxdbConstants;
import com.yejiarong.influxdb.api.bean.request.SavePointRequest;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * @author YeJR
 */
@Slf4j
@Service
public class MQConsumerService {

    private final InfluxDBClient influxDBClient;

    public MQConsumerService(InfluxDBClient influxDBClient) {
        this.influxDBClient = influxDBClient;
    }

    @SneakyThrows
    @RabbitListener(queues = "weather_01")
    public void weatherConsumer01(Channel channel, Message message) {
        SavePointRequest savePointRequest = JSONObject.parseObject(new String(message.getBody()), SavePointRequest.class);
        Date date = DateUtils.parseDate(savePointRequest.getTime());
        Point point = Point
                .measurement(savePointRequest.getMetric())
                .addTags(savePointRequest.getTags())
                .addFields(convertPoints(savePointRequest.getPoints()))
                .time(date.getTime(), WritePrecision.MS);

        WriteApiBlocking writeApi = influxDBClient.getWriteApiBlocking();
        writeApi.writePoint(InfluxdbConstants.BUCKET_WEATHER, InfluxdbConstants.ORG_SHANGHAI, point);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    private Map<String, Object> convertPoints(Map<String, Object> points) {
        Map<String, Object> result = new HashMap<>();
        for (Map.Entry<String, Object> entry : points.entrySet()) {
            if (Objects.nonNull(entry.getValue())) {
                try {
                    if (entry.getValue().toString().contains(".")) {
                        result.put(entry.getKey(), Float.parseFloat(entry.getValue().toString()));
                    } else {
                        result.put(entry.getKey(), Integer.parseInt(entry.getValue().toString()));
                    }
                } catch (NumberFormatException e) {
                    result.put(entry.getKey(), entry.getValue());
                }
            }
        }
        return result;
    }
}
